package com.heima.article.job;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.heima.article.service.HotArticleService;
import com.heima.common.constants.article.HotArticleConstants;
import com.heima.model.mess.app.ArticleVisitStreamMess;
import com.heima.model.mess.app.UpdateArticleMess;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.BinaryOperator;
import java.util.stream.Collectors;

@Component
@Slf4j
public class updateHotArticleJob {

    @Autowired
    StringRedisTemplate redisTemplate;
    @Autowired
    HotArticleService hotArticleService;

    @XxlJob("updateHotArticleJob")
    public ReturnT handel(String params) {
        //1 . 先获取redis中的行为数据
        List<UpdateArticleMess> articleMessList = getUpdateArticleMesses();

        if (CollectionUtils.isEmpty(articleMessList)) {
            log.info("太冷清了， 没人访问我们的数据");
            return ReturnT.SUCCESS;
        }

        //2 . 按照文章id对行为数据进行分类 统计出聚合数据
       List<ArticleVisitStreamMess> waitUpdateArticleScore =  getWaitUpdateArticleScore(articleMessList);
        if (CollectionUtils.isEmpty(waitUpdateArticleScore)) {
            log.info("太冷清了， 没人访问我们的数据");
            return ReturnT.SUCCESS;
        }

        //3 .修改文章的分值
        //3.1 遍历统计好的聚合数据，调用方法修改每个文章的热度值
        waitUpdateArticleScore.forEach((hotArticleService::updateApArticle));



        return ReturnT.SUCCESS;
    }






    /**
     * 按照文章id对行为数据进行分类 统计出聚合数据
     * @param articleMessList
     * @return
     */
    private List<ArticleVisitStreamMess> getWaitUpdateArticleScore(List<UpdateArticleMess> articleMessList) {
    List<ArticleVisitStreamMess> waitUpdateArticleScoreData = new ArrayList<>();

        //1 . 按照文章id进行分组 ， key： 文章id  value： 文章行为数据list
        Map<Long, List<UpdateArticleMess>> meeByGroup =
                articleMessList.stream()
                        .collect(Collectors.groupingBy(UpdateArticleMess::getArticleId));



        //2 . 遍历分组 ， 对每个分组进行聚合运算  每个分组统计出一个 ArticleVisitStreamMess对象

        meeByGroup.forEach((articleId , messList) -> {


            Optional<ArticleVisitStreamMess> reduce = messList.stream()
                    .map(updatemess -> {
                        //将每一条行为数据都转化为ArticleVisitStreamMess聚合对象
                        ArticleVisitStreamMess visitStreamMess = new ArticleVisitStreamMess();
                        visitStreamMess.setArticleId(articleId);//设置文章Id
                        switch (updatemess.getType()) {
                            case LIKES:
                                visitStreamMess.setLike(updatemess.getAdd());
                                break;
                            case VIEWS:
                                visitStreamMess.setView(updatemess.getAdd());
                                break;
                            case COMMENT:
                                visitStreamMess.setComment(updatemess.getAdd());
                                break;
                            case COLLECTION:
                                visitStreamMess.setCollect(updatemess.getAdd());
                                break;
                        }
                        return visitStreamMess;
                        /**
                         * 将流中的数据进行两两运算
                         */
                    }).reduce(new BinaryOperator<ArticleVisitStreamMess>() {
                        @Override
                        public ArticleVisitStreamMess apply(ArticleVisitStreamMess a1, ArticleVisitStreamMess a2) {
                            a1.setLike(a1.getLike() + a2.getLike());
                            a1.setLike(a1.getView() + a2.getView());
                            a1.setLike(a1.getCollect() + a2.getCollect());
                            a1.setLike(a1.getComment() + a2.getComment());
                            return a1;
                        }
                    });
            waitUpdateArticleScoreData.add(reduce.get());
        });

        return waitUpdateArticleScoreData;
    }


    /**
     * 获取redis中的行为数据
     *
     * @return
     */
    private List<UpdateArticleMess> getUpdateArticleMesses() {
        //1.1得到size
        ListOperations<String, String> opsForList = redisTemplate.opsForList();
        Long size = opsForList.size(HotArticleConstants.HOT_ARTICLE_SCORE_BEHAVIOR_LIST);

        //1.2 然后通过lrange（0~size-1） ltrim（ size  -1）
        List result = redisTemplate.executePipelined(new RedisCallback<UpdateArticleMess>() {
            @Override
            public UpdateArticleMess doInRedis(RedisConnection redisConnection) throws DataAccessException {
                redisConnection.openPipeline();//开启管道命令
                // lrange（0~size-1） ltrim（ size  -1）
                redisConnection.lRange(HotArticleConstants.HOT_ARTICLE_SCORE_BEHAVIOR_LIST.getBytes(), 0, size - 1);
                redisConnection.lTrim(HotArticleConstants.HOT_ARTICLE_SCORE_BEHAVIOR_LIST.getBytes(), size, -1);
                return null;
            }
        });
        if (CollectionUtils.isNotEmpty(result)) {
            List<String> list = (List<String>) result.get(0);
            return list.stream().map(jsonStr-> JSON.parseObject(jsonStr,UpdateArticleMess.class)).collect(Collectors.toList());
        }
        return null;
    }
}
